package com.atguigu.chapter06;
import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * Author: Pepsi
 * Date: 2023/7/31
 * Desc:
 */
public class Flink05_Project_Order {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        SingleOutputStreamOperator<OrderEvent> orderEventStream = env
                .readTextFile("input/OrderLog.csv")
                // 过滤出 pay 事件
                .filter(line -> {
                    String[] datas = line.split(",");
                    return "pay".equals(datas[1]);
                })
                // 封装成对象
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new OrderEvent(
                                Long.valueOf(datas[0]),
                                datas[1],
                                datas[2],
                                Long.valueOf(datas[3])
                        );
                    }
                });

        SingleOutputStreamOperator<TxEvent> txEventStream = env
                .readTextFile("input/ReceiptLog.csv")
                // 封装成对象
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new TxEvent(
                                datas[0],
                                datas[1],
                                Long.valueOf(datas[2])
                        );
                    }
                });

        // 连接两个不同类型的流
        orderEventStream.connect(txEventStream)
                // 将连接后的流，用交易流水id分组
                .keyBy(new KeySelector<OrderEvent, String>() {
                    @Override
                    public String getKey(OrderEvent value) throws Exception {
                        return value.getTxId();
                    }
                }, new KeySelector<TxEvent, String>() {
                    @Override
                    public String getKey(TxEvent value) throws Exception {
                        return value.getTxId();
                    }
                })
                // 用keyed这个能获得的信息多一些，写起来简单
                // 泛型的含义分别是key的类型，第一个流输入的类型，第二个流输入的类型，输出类型
                .process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {

                    // 存支付订单的支付流水号，这两个流来的先后顺序不一定，如果先来的找不到另一个那就先存起来
                    // key是流水号，value是事件对象
                    Map<String,OrderEvent> orderEventMap = new HashMap<>();
                    // 存支付平台收据日志的流水号
                    Map<String,TxEvent> txEventMap = new HashMap<>();

                    @Override
                    public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                        // 获取到订单事件的流水号，然后去收据日志集合里面找有没有这个流水号
                        TxEvent txEvent = txEventMap.get(value.getTxId());
                        if (txEvent != null){
                            // 如果有这个流水号，说明到账了，将相关信息放到流中
                            out.collect("订单: " + value.getOrderId() + " 对账成功!!!!");
                        }else{
                            // 如果没有找到，那就将流水号，和对象放到集合中
                            orderEventMap.put(ctx.getCurrentKey(),value);
                        }
                    }

                    @Override
                    public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                        // 拿到流水号，去订单集合里面找
                        OrderEvent orderEvent = orderEventMap.get(ctx.getCurrentKey());
                        if (orderEvent !=null){
                            // 找到了，将相关信息放到流中
                            out.collect("订单: " + orderEvent.getOrderId() + " 对账成功!!!!");
                        }else {
                            // 没有找到，存起来
                            txEventMap.put(ctx.getCurrentKey(), value);
                        }
                    }
                })
                // 将流中的信息打印出来
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
